package com.study.v115;

import com.study.utils.JsonHelper;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.apache.flink.util.Collector;

/**
 * @author sjw
 * @className FraudDetectionJob
 * @date 2022/7/7 14:07
 * @description: TODO
 */
public class FraudDetectionJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //这些配置建议可以放在配置文件中，然后通过 parameterTool 来获取对应的参数值
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("192.168.10.200").setVirtualHost("OuterHost")
                .setPort(5672).setUserName("root").setPassword("root")
                .build();

        DataStreamSource<String> dataStreamSource = env.addSource(new RMQSource<>(connectionConfig,
                "q.study.flink",
                true,
                new SimpleStringSchema()))
                .setParallelism(1);

        DataStream<User> users = dataStreamSource.process(new ProcessFunction<String, User>() {
            @Override
            public void processElement(String value, Context ctx, Collector<User> out) throws Exception {
                User user = JsonHelper.jsonToObject(value, User.class);
                out.collect(user);
            }
        }).name("");


        users.keyBy(User::getId).process(new KeyedProcessFunction<Integer, User, User>() {
            transient ValueState<Boolean> flagState;

            @Override
            public void open(Configuration parameters) {
                ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>(
                        "flag",
                        Types.BOOLEAN);
                flagState = getRuntimeContext().getState(flagDescriptor);
            }

            @Override
            public void processElement(User value, Context ctx, Collector<User> out) throws Exception {
                Boolean flag = flagState.value();
                if(value.id==1){
                    if(flag==null){
                        System.out.println("null" +JsonHelper.objectToJson(value));
                        flagState.update(true);
                    }else if(flag){
                        System.out.println("clear" +JsonHelper.objectToJson(value));
                        out.collect(value);
                        flagState.clear();
                    }
                }
            }
        }).addSink(new SinkFunction<User>() {
            @Override
            public void invoke(User value, Context context) throws Exception {
                System.out.println(value.id + " " + value.name);
            }
        });

        env.execute();
    }

    private static class User {
        private Integer id;
        private String name;

        public Integer getId() {
            return id;
        }

        public void setId(Integer id) {
            this.id = id;
        }

        public void setName(String name) {
            this.name = name;
        }
    }

}
